package homework5

import java.util.UUID
import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import homework5.Common._
import com.typesafe.config.ConfigFactory

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

/**
 * scala很简洁，没有Java废话，但是同样的阅读起来也是有一些困难
 * 有的地方没有那么明显的语意
 *
 * @param masterHost
 * @param masterPort
 */
class SparkWorker(masterHost: String, masterPort: Int) extends Actor {

  // 定义ActorSelection对象，这里的是一个spackMater，给spackMater发消息
  var masterProxy: ActorSelection = _


  // 预启动，初始化spackMater，通过网络初始化，很像dubbo初始化提供者代理
  override def preStart(): Unit = {
    super.preStart()
    masterProxy = context.actorSelection(s"akka.tcp://SparkMaster@${masterHost}:${masterPort}/user/SparkMaster")
  }

  // 消息id
  val id = UUID.randomUUID().toString

  override def receive: Receive = {
    // 收到开始消息
    case StartMessage(msg) =>
      println(s"Spark Worker start. $msg")
      // worker启动之后，向master注册，即给master发送注册消息
      masterProxy ! RegisterWorkInfo(id, 8, 8 * 1024)
    // 收到注册成功消息
    case RegisteredWorkerInfo => {
      println(s"收到sparkMaster回复消息,$id 注册成功")
      // worker注册完成后，启动定时任务，向master发送心跳,即给自己发送心跳命令消息
      context.system.scheduler.schedule(0 millis, 3000 millis, self, SendHeartBeat)
    }
    // 收到心跳命令
    case SendHeartBeat => {
      println(s"worker[${id}]发送心跳！")
      masterProxy ! HeartBeat(id)
    }
  }
}

object SparkWorker extends App {
  // acotor配置信息
  val (masterHost, masterPort, workerHost, workerPort) = ("127.0.0.1", 10010, "127.0.0.1", 10001)
  val config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$workerHost
       |akka.remote.netty.tcp.port=$workerPort
      """.stripMargin
  )

  // 初始化SparkWorker
  val sparkWorkerSystem = ActorSystem("SparkWorker", config)
  private val sparkActor: ActorRef = sparkWorkerSystem.actorOf(Props(new SparkWorker(masterHost, masterPort)), "SparkWorker-01")

  // 给自己发送开始消息
  sparkActor ! StartMessage(s"host:$workerHost port:$workerPort")
}
